package com.zhangyang.flink.cdc;


import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.api.common.serialization.SerializationSchema;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.Map;

import static org.apache.flink.api.java.typeutils.TypeExtractor.getForClass;

/**
 * @author Administrator
 * @Title:
 * @Description: 请写注释类
 * @date 2022/6/17
 */
public class MapDeserializationSchema implements DeserializationSchema<Map>, SerializationSchema<Map<String,Object>> {

    private static final Logger LOG = LoggerFactory.getLogger(MapDeserializationSchema.class);

    private static final ObjectMapper mapper = new ObjectMapper();

    private String dataType;

    public MapDeserializationSchema() {
    }

    public MapDeserializationSchema(String dataType) {
        this.dataType = dataType;
    }

    @Override
    public boolean isEndOfStream(Map nextElement) {
        return false;
    }

    @Override
    public TypeInformation<Map> getProducedType() {
        return getForClass(Map.class);
    }

    @Override
    public Map deserialize(byte[] message) throws IOException {
        Map collectionLogInfo = null;
        try {

                collectionLogInfo = mapper.readValue(new String(message), Map.class);

        } catch (Exception e) {
            LOG.error("无法反序列化成对象[{}]", new String(message));
        }
        return collectionLogInfo;
    }

    @Override
    public byte[] serialize(Map<String, Object> element) {
        byte[] bytes = new byte[0];
        try {

                bytes = mapper.writeValueAsBytes(element);

        } catch (Exception e) {
            LOG.error("无法序列化成字节数组[{}]", e);
        }
        return bytes;
    }
}
